package com.dongql.reactor;

import java.util.concurrent.Flow.Subscriber;
import java.util.concurrent.Flow.Subscription;
import java.util.function.Consumer;

class SampleSubscriber<T> implements Subscriber<T> {

    final Consumer<? super T> consumer;
    Subscription subscription;
    final long bufferSize;
    long count;

    SampleSubscriber(long bufferSize, Consumer<? super T> consumer) {
        this.bufferSize = bufferSize;
        this.consumer = consumer;
    }

    @Override
    public void onSubscribe(Subscription subscription) {
        long initialRequestSize = bufferSize;
        // re-request when half consumed
        count = bufferSize - bufferSize / 2;
        (this.subscription = subscription).request(initialRequestSize);
    }

    @Override
    public void onNext(T item) {
        if (--count <= 0) {
            subscription.request(count = bufferSize - bufferSize / 2);
        }
        consumer.accept(item);
    }

    @Override
    public void onError(Throwable ex) {
        ex.printStackTrace();
    }

    @Override
    public void onComplete() {
    }
}